package com.cloud.rocketmq.controller;

import com.cloud.rocketmq.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/rocket")
@Slf4j
public class RocketMqController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 普通消息
    @RequestMapping("/sendSimple")
    public void sendSimple() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        rocketMQTemplate.convertAndSend("rocketMq", user);
    }

    // 同步消息
    @RequestMapping("/syncSend")
    public void syncSend() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        SendResult rocketMq = rocketMQTemplate.syncSend("rocketMq", user);
        System.out.println(rocketMq);
    }

    // 异步消息
    @RequestMapping("/asyncSend")
    public void asyncSend() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        rocketMQTemplate.asyncSend("rocketMq", user, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
                System.out.println(sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败");
                System.out.println(throwable);

            }
        });
    }

    // 单向消息
    @RequestMapping("/sendOneWay")
    public void sendOneWay() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        rocketMQTemplate.sendOneWay("rocketMq", user);
    }

    // 延时消息
    //RocketMQ不支持任意时间的延时，只支持以下几个固定的延时等级
    //1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    @RequestMapping("/sendTimeSync")
    public void sendTimeSync() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        Message<User> userMessageBuilder = MessageBuilder.withPayload(user).build();
        //3代指10s
        SendResult rocketMq = rocketMQTemplate.syncSend("rocketMq", userMessageBuilder, 3000, 3);
    }

    // 顺序消息
    @RequestMapping("/sendOrder")
    public void sendOrder() {
        User user = new User();
        user.setName("张三");
        user.setAge(15);
        // 顺序消息,发送数据可能导致数据顺序改变
//      //第三个参数是order ,他的作用是会根据他的hash值计算发送到哪一个队列
        rocketMQTemplate.syncSendOrderly("rocketOrder", "1", "order");
        rocketMQTemplate.syncSendOrderly("rocketOrder", "2", "order");
        rocketMQTemplate.syncSendOrderly("rocketOrder", "3", "order");
    }

    // 指定tag消费
    //消费端，通过selectorExpression = “TAG1 || TAG2”,selectorType = SelectorType.TAG，指定需要消费的TAG
    // 只有TAG1与TAG2能接收到
    @RequestMapping("/sendTag")
    public void sendTag() {
        // 构造消息1
        Message msg1 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG01").build();
        // 构造消息2
        Message msg2 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG02").build();
        // 构造消息3
        Message msg3 = MessageBuilder.withPayload("rocketmq过滤消息测试-TAG03").build();

        rocketMQTemplate.convertAndSend("rocketTag" + ":" + "TAG1", msg1);
        rocketMQTemplate.convertAndSend("rocketTag" + ":" + "TAG2", msg2);
        rocketMQTemplate.convertAndSend("rocketTag" + ":" + "TAG3", msg3);
    }

    @RequestMapping("/sendTransactionMessage")
    public void sendTransactionMessage() {
        Message<String> userMessageBuilder = MessageBuilder.withPayload("事务").build();
//        myTransactionGroup要和@RocketMQTransactionListener(txProducerGroup = "myTransactionGroup")定义的Group一致
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction("productGroup", "rocketMq", userMessageBuilder, null);
        String transactionId = transactionSendResult.getTransactionId();
        String status = transactionSendResult.getSendStatus().name();
        log.info("发送消息成功 transactionId={} status={} ", transactionId, status);
    }

}
